import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 马屎咖啡罐
 *
 * @author <achao1441470436@gmail.com>
 * @since 2021/11/4 0004 22:25
 */
public class HorseShitCafeJob {
    public static void main(String[] args) throws Exception {
        // 看了这篇文章 https://mp.weixin.qq.com/s/PKAXqR-kDeWd0Vvq891NEA 试着实现其中的案例
        // 创业不易，需要绞尽脑汁吸引顾客。于是你搞了一个促销活动，一杯咖啡积1分，每个顾客只要积够10分，就白送一次骑马体验。
        // 你店里卖的东西种类越来越多：10种咖啡，8种点心，还有6种正餐。
        // 与此同时，你的促销活动也变得更复杂：一份咖啡积1分，一份点心积0.5分，一份正餐积2分。
        // 一周内积够10分才送骑马，七天之前的过期积分就作废。

        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 数据源，此处我使用Iterator模拟无限的用户
        DataStream<Order> orders = env.addSource(new OrderSource()).name("orders");

        // 实际计算，此处就是实时计算员
        DataStream<HorseTicket> alerts = orders.keyBy(Order::getUsername).process(new Calculator()).name("calculator");

        // 添加计算后的下游，此处就是马场
        alerts.addSink(new Racecourse()).name("racecourse");

        // 执行
        env.execute("Horse Shit Cafe");
    }
}
